/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/

package com.skjegstad.mist2.modules.udp;

import com.skjegstad.mist2.datatypes.Acknowledgement;
import com.skjegstad.mist2.datatypes.BloomfilterAcknowledgment;
import com.skjegstad.mist2.datatypes.ListAcknowledgement;
import com.skjegstad.mist2.datatypes.NodeIdentifier;
import com.skjegstad.mist2.datatypes.TopicIdentifier;
import com.skjegstad.mist2.datatypes.TopicMap;
import com.skjegstad.mist2.datatypes.TopicMapEntry;
import com.skjegstad.mist2.exceptions.MistException;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.LinkedList;
import java.util.Map;

/**
 * Class describing subscribe messages.
 *
 * @author Magnus Skjegstad
 */
public class BeaconMessage implements Externalizable {
    private TopicMap subscribeToTopics;
    private Acknowledgement acknowledgements;
    private NodeIdentifier nodeIdentifier;
    private int maximumAge;
    private int contactPort;

    public BeaconMessage(NodeIdentifier nodeIdentifier,
            TopicMap subscribeToTopics,
            int contactPort,
            int maximumAge,
            Acknowledgement ack) {
        super();
        this.subscribeToTopics = subscribeToTopics;
        this.maximumAge = maximumAge;
        this.acknowledgements = ack;
        this.nodeIdentifier = nodeIdentifier;
        this.contactPort = contactPort;
    }

    public BeaconMessage() {        
    }

    public NodeIdentifier getNodeIdentifier() {
        return nodeIdentifier;
    }


    public Acknowledgement getAcknowledgements() {
        return acknowledgements;
    }

    public void setAcknowledgements(BloomfilterAcknowledgment acknowledgements) {
        this.acknowledgements = acknowledgements;
    }

    public TopicMap getSubscribeToTopics() {
        if (subscribeToTopics == null) {
            subscribeToTopics = new TopicMap();
        }
        return subscribeToTopics;
    }

    public void setSubscribeToTopics(TopicMap subscribeToTopics) {
        this.subscribeToTopics = subscribeToTopics;
    }

    public int getMaximumAge() {
        return maximumAge;
    }

    public void setMaximumAge(int maximumAge) {
        this.maximumAge = maximumAge;
    }

    public int getContactPort() {
        return contactPort;
    }

    @Override
    public String toString() {
        String s = new String();
        
        //s += "sourceNodeId:" + this.getSourceId() + " ";
       // s += "ttl: " + this.getTtl() + " ";
        s += "subscribeToTopics:" + this.getSubscribeToTopics() + " ";
        s += "acknowledgements:" + this.getAcknowledgements() + " ";
        s += "maximumAge:" + this.getMaximumAge() + " ";
        
        return s;
    }

    public void writeExternal(ObjectOutput out) throws IOException {
        byte flags = 0;
        if (this.subscribeToTopics != null && !this.subscribeToTopics.isEmpty())
            flags |= 0x01;
        if (this.acknowledgements != null)
            flags |= 0x02;
        if (this.contactPort > 0) {
            flags |= 0x04;
        }
        if (this.acknowledgements instanceof ListAcknowledgement)
            flags |= 0x08;
        
        out.writeByte(flags);

        out.writeLong(nodeIdentifier.getNodeIdentifier());

        out.writeInt(maximumAge);

        if (this.subscribeToTopics != null && !this.subscribeToTopics.isEmpty()) {
            out.writeInt(subscribeToTopics.keySet().size());
            for (Map.Entry<TopicIdentifier, TopicMapEntry> e : subscribeToTopics.entrySet()) {
                out.writeShort(e.getKey().getHashes().size());
                for (int i : e.getKey().getHashes())
                    out.writeInt(i);
                out.writeInt(e.getValue().getTtl());
                out.writeLong(e.getValue().getChecksum());
            }
        }
        if (this.acknowledgements != null) {
            byte[] data = this.getAcknowledgements().getData();
            out.writeInt(data.length); // length in bytes
            out.writeInt(this.acknowledgements.getLengthInBits()); // length in bits
            out.writeByte(this.acknowledgements.getK());
            out.write(data); // bytes            
        }
        if (this.contactPort > 0)
            out.writeInt(contactPort);
    }

    public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
        byte flags = in.readByte();

        this.nodeIdentifier = new NodeIdentifier(in.readLong());

        this.maximumAge = in.readInt();

        if ((flags & 0x01) == 0x01) {// subscribeToTopics
            int totalTopics = in.readInt();
            this.subscribeToTopics = new TopicMap();
            for (int i = 0; i < totalTopics; i++) {
                
                short len = in.readShort();
                LinkedList<Integer> hashes = new LinkedList();
                for (int j = 0; j < len; j++)
                    hashes.add(in.readInt());
                TopicIdentifier t = new TopicIdentifier(hashes);

                int ttl = in.readInt();
                long checksum = in.readLong();
                this.subscribeToTopics.put(t, new TopicMapEntry(ttl, checksum));
            }
        }
        if ((flags & 0x02) == 0x02) { // acknowledgements
            int len = in.readInt();
            int lenInBits = in.readInt();
            byte k = in.readByte();
            byte[] data = new byte[len];

            int offset = 0;
            int read = 0;
            while (read > -1 && offset < data.length) {
                read = in.read(data, offset, data.length-offset);
                offset = offset + read;
            }
            try {
                if ((flags & 0x08) == 0x08) // list ack
                    this.acknowledgements = new ListAcknowledgement(data);
                else
                    this.acknowledgements = new BloomfilterAcknowledgment(data, lenInBits, k);
            } catch (MistException ex) {
                throw new IOException("Unable to deserialize bloomfilter"); // Cause is not given to stay compatible with 1.5
            }
        }
        if ((flags & 0x04) == 0x04) { // contact port
            contactPort = in.readInt();
        }
    }
}
